import { Inject } from '@midwayjs/decorator';
import { BaseService, CoolCommException } from '@cool-midway/core';
import { InjectEntityModel } from '@midwayjs/typeorm';
import { Equal, Repository } from 'typeorm';
import { TaskJobEntity } from '../entity/job';
import { TaskInitService } from './init';
import { Provide, Scope, ScopeEnum } from '@midwayjs/core';
import { TaskTopicEntity } from '../entity/topic';
import * as _ from 'lodash';
import { TaskRecordEntity } from '../entity/record';
import * as moment from 'moment';

/**
 * 任务作业
 */
@Scope(ScopeEnum.Request, { allowDowngrade: true })
@Provide()
export class TaskJobService extends BaseService {
  @InjectEntityModel(TaskJobEntity)
  taskJobEntity: Repository<TaskJobEntity>;

  @InjectEntityModel(TaskTopicEntity)
  taskTopicEntity: Repository<TaskTopicEntity>;

  @InjectEntityModel(TaskRecordEntity)
  taskRecordEntity: Repository<TaskRecordEntity>;

  @Inject()
  taskInitService: TaskInitService;

  /**
   * 修改之前
   * @param data
   * @param type
   */
  async modifyBefore(data: any, type: 'delete' | 'update' | 'add') {
    if (type == 'add' || type == 'update') {
      // 判断connectOptions 如果不为空，需要判断是否是json格式的字符串
      if (data.body) {
        try {
          let result = JSON.parse(data.body);
          if (typeof result !== 'object' || result === null) {
            throw new Error();
          }
        } catch (error) {
          throw new CoolCommException('body需要是json格式的字符串');
        }
      } else {
        data.body = null;
      }
    }
  }

  /**
   * 新增|删除
   * @param param
   */
  async addOrUpdate(param: TaskJobEntity) {
    await super.addOrUpdate(param);
    const info = await this.taskJobEntity.findOneBy({
      id: Equal(param.id),
    });
    await this.save(info, param.appId);
  }

  /**
   * 结果
   * @param recordId
   * @param status
   * @param result
   */
  async result(recordId: number, status: number, result: string) {
    const record = await this.taskRecordEntity.findOneBy({ id: recordId });
    if (!record) {
      throw new CoolCommException('记录不存在');
    }
    await this.taskRecordEntity.update(recordId, {
      status,
      result,
      resultTime: new Date(),
    });
  }

  /**
   * 通过ID启动任务
   * @param id
   */
  async startById(id: number) {
    const info = await this.taskJobEntity.findOneBy({ id });
    await this.start(info.jobId, info.topic, info.appId);
  }

  /**
   * 启动任务
   * @param jobId
   * @param topic
   * @param appId
   */
  async start(jobId: string, topic: string, appId: number) {
    const job = await this.taskJobEntity.findOneBy({
      jobId: Equal(jobId),
      topic: Equal(topic),
      appId,
    });
    if (!job) {
      throw new CoolCommException('任务不存在');
    }
    job.status = job.type == 0 ? 1 : 0;
    await this.save(job, appId);
  }

  /**
   * 结束bullmq任务
   * @param jobId
   * @param topic
   * @param appId
   */
  async bullRemove(jobId: string, topic: string, appId: number) {
    const bullJobId = `${appId}@${topic}@${jobId}`;
    const queue = await this.taskInitService.getQueue(appId, topic);
    if (!queue) return;
    await queue.remove(bullJobId);
    const result = await queue.getRepeatableJobs();
    const job = _.find(result, { id: bullJobId });
    if (job) {
      await queue.removeRepeatableByKey(job.key);
    }
  }

  /**
   * 通过Id停止任务
   * @param id
   */
  async stopById(id: number) {
    const info = await this.taskJobEntity.findOneBy({ id });
    await this.stop(info.jobId, info.topic, info.appId);
  }

  /**
   * 停止任务
   * @param jobId
   * @param topic
   * @param appInfo
   */
  async stop(jobId: string, topic: string, appId: number) {
    await this.bullRemove(jobId, topic, appId);
    await this.taskJobEntity.update(
      {
        jobId: Equal(jobId),
        topic: Equal(topic),
        appId,
      },
      {
        status: 3,
      }
    );
  }

  /**
   * 通过任务ID移除任务
   * @param id
   */
  async removeById(id: number) {
    const info = await this.taskJobEntity.findOneBy({ id });
    await this.remove(info.jobId, info.topic, info.appId);
  }

  /**
   * 移除任务
   * @param jobId
   * @param topic
   * @param appInfo
   */
  async remove(jobId: string, topic: string, appId: number) {
    await this.bullRemove(jobId, topic, appId);
    await this.taskJobEntity.delete({
      jobId: Equal(jobId),
      topic: Equal(topic),
      appId: appId,
    });
  }

  /**
   * 任务详情
   * @param jobId
   * @param topic
   * @param appInfo
   * @returns
   */
  async detail(jobId: string, topic: string, appId: number) {
    const job = await this.taskJobEntity.findOneBy({
      jobId: Equal(jobId),
      topic: Equal(topic),
      appId: appId,
    });
    return job;
  }

  /**
   * 更新下一次执行时间
   * @param id
   */
  async updateNextRunTime(id: number) {
    let nextRunTime;
    const jobInfo = await this.taskJobEntity.findOneBy({ id });
    const jobId = `${jobInfo.appId}@${jobInfo.topic}@${jobInfo.jobId}`;
    const queue = await this.taskInitService.getQueue(
      jobInfo.appId,
      jobInfo.topic
    );
    if (!queue) {
      return;
    }
    // 重复任务
    if (jobInfo.type == 0) {
      // cron
      if (jobInfo.cron) {
        const result = await queue.getRepeatableJobs();
        const task = _.find(result, { id: jobId });
        nextRunTime = new Date(task.next);
      }
      if (jobInfo.every) {
        nextRunTime = new Date(new Date().getTime() + jobInfo.every);
      }
    }
    // 延时任务
    if (jobInfo.type == 1) {
      // 当前时间+延时时间jobInfo.delay 是毫秒， 用moment
      nextRunTime = moment().add(jobInfo.delay, 'ms').toDate();
    }
    await this.taskJobEntity.update(id, { nextRunTime });
    return { nextRunTime };
  }

  /**
   * 创建任务
   * @param job
   * @param appInfo
   */
  async save(job: TaskJobEntity, appId: number) {
    job.appId = appId;
    await this.check(job, appId);
    const jobId = `${appId}@${job.topic}@${job.jobId}`;
    const queue = await this.taskInitService.getQueue(appId, job.topic);
    // 先移除任务，如果存在的话
    await this.bullRemove(job.jobId, job.topic, appId);
    // 重复任务
    if (job.type == 0) {
      job.status = 1;
      const repeat = {
        jobId,
        pattern: job.cron,
        every: job.every,
        startDate: job.startTime,
      };
      delete job.nextRunTime;
      queue.add(job.name, job, { jobId, repeat });
    }
    // 延时任务
    if (job.type == 1) {
      job.status = 0;
      queue.add(job.name, job, { jobId, delay: job.delay });
    }
    const old = await this.taskJobEntity.findOneBy({
      jobId: Equal(job.jobId),
      topic: Equal(job.topic),
      appId,
    });
    if (old) {
      job.id = old.id;
    }
    await this.taskJobEntity.save(job);
    // 更新下一次执行时间
    return await this.updateNextRunTime(job.id);
  }

  /**
   * 检查数据
   * @param job
   * @param appInfo
   */
  async check(job: TaskJobEntity, appId: number) {
    if (!job.topic) throw new CoolCommException('Topic不能为空');
    const topic = await this.taskTopicEntity.findOneBy({
      name: Equal(job.topic),
      status: 1,
      appId,
    });
    if (!topic) throw new CoolCommException('Topic不存在或被禁用');
    const queue = await this.taskInitService.getQueue(appId, job.topic);
    if (!queue) throw new CoolCommException('Queue不存在');
    // 重复任务
    if (job.type == 0) {
      if (job.cron) {
        job.every = null;
      } else {
        job.cron = null;
      }
      if (!job.cron && !job.every) {
        throw new CoolCommException('cron和every不能同时为空');
      }
    }
    // 延时任务
    if (job.type == 1) {
      if (!job.delay) {
        throw new CoolCommException('delay不能为空');
      }
    }
  }
}
